<?php
/**
 * @brief RabbitMQ消息类
 */
class Amqp {
    protected $exchangeName = '';    // 交换机名称
    protected $routeName    = '';    // 路由键名称
    protected $queueName    = '';    // 队列名称
    protected $conn         = null;  // 连接
    protected $channel      = null;  // 通道
    protected $exchange     = null;  // 交换机
    protected $queue        = null;  // 队列

    /**
     * 初始化配置
     * 
     * @param string $queueName 队列名称
     * 
     * @return void
     */
    public function __construct($queueName) {
        if (!extension_loaded('amqp')) {
            throw new \Exception('The amqp extension must be loaded before use');
        }

        if (!is_string($queueName) || empty($queueName)) {
			throw new \Exception('Queue name must be specified in a string');
		}
        $queueName = trim($queueName);
        list($exchangeName, $routeName) = explode('_', $queueName, 2);
        $this->exchangeName = $exchangeName;
        $this->routeName = $routeName;
        $this->queueName = $queueName;
        
        $config = Yaf_Registry::get('Config');
        // 配置信息
        $args = [
            'host'     => $config->amqp->host,
            'port'     => $config->amqp->port,
            'login'    => $config->amqp->user,
            'password' => $config->amqp->password,
            'vhost'    => $config->amqp->vhost
        ];
        // 创建连接
        $this->conn = new \AMQPConnection($args);
        if (!$this->conn->connect()) {
            throw new \Exception('Cannot connect to the RabbitMQ broker');
        }
        // 在连接内创建一个通道
        $this->channel = new \AMQPChannel($this->conn);
        // 创建交换机
        $this->exchange = new \AMQPExchange($this->channel);
        $this->exchange->setName($this->exchangeName);
        // 设置交换机类型
        // AMQP_EX_TYPE_DIRECT:直连交换机
        // AMQP_EX_TYPE_FANOUT:扇形交换机
        // AMQP_EX_TYPE_HEADERS:头交换机
        // AMQP_EX_TYPE_TOPIC:主题交换机
        $this->exchange->setType(AMQP_EX_TYPE_DIRECT);
        // 设置交换机持久化
        $this->exchange->setFlags(AMQP_DURABLE);
        // 声明交换机
        $this->exchange->declareExchange();
    }

    /**
     * 发送一条消息
     * 
     * @param string|array $data 消息数据
     * 
     * @return boolean
     */
    public function sendMsg($data) {
        if (is_array($data)) {
            $data = json_encode($data);
        }
        return $this->exchange->publish($data, $this->routeName) ? true : false;
    }

    /**
     * 初始化队列
     * 
     * @return void
     */
    private function initQueue() {
        // 创建队列
        $this->queue = new \AMQPQueue($this->channel);
        $this->queue->setName($this->queueName);
        // 设置队列持久化
        $this->queue->setFlags(AMQP_DURABLE);
        // 声明消息队列
        $this->queue->declareQueue();
        // 绑定交换机与队列，并指定路由键
        $this->queue->bind($this->exchangeName, $this->routeName);
    }

    /**
     * 从当前队列中取出一条消息
     * 
     * @return mixed
     */
    public function getMsg() {
        $this->queue || $this->initQueue();
        return $this->queue->get(AMQP_AUTOACK)->getBody();
    }

    /**
     * 消费者阻塞接收消息(限CLI模式下使用)
     * 
     * @return void
     */
    public function receiveMsg() {
        // 引入消息队列的回调函数
        Yaf_Loader::import(APP_PATH . '/application/callbacks/Amqp.php');

        $this->queue || $this->initQueue();
        $queueName = $this->queueName;
        
        // 设置消息回调
        $this->queue->consume(function($envelope, $queue) use($queueName) {
            // 消息处理
            $func = 'amqp_' . $queueName;
            $func($envelope->getBody());
            // 显式确认，队列收到消费者显式确认后，会自动删除该消息
            $queue->ack($envelope->getDeliveryTag());
        });
    }

    /**
     * 删除当前队列
     * 
     * @return boolean
     */
    public function deleteQueue() {
        $this->queue || $this->initQueue();
        return $this->queue->delete();
    }

    /**
     * 清空当前队列中的消息
     * 
     * @return boolean
     */
    public function clearQueue() {
        $this->queue || $this->initQueue();
        return $this->queue->purge();
    }

    /**
     * 开始事务
     * 
     * @return void
     */
    public function startTransaction() {
        $this->channel->startTransaction();
    }

    /**
     * 提交事务
     * 
     * @return void
     */
    public function commitTransaction() {
        $this->channel->commitTransaction();
    }

    /**
     * 回滚事务
     * 
     * @return void
     */
    public function rollbackTransaction() {
        $this->channel->rollbackTransaction();
    }
    
    /**
     * 关闭连接
     * 
     * @return void
     */
    public function close() {
        $this->conn->disconnect();
    }
}
